//package com.gukun.springboot.tdengine.service.topic;
//
//import cn.hutool.core.date.DateTime;
//import cn.hutool.core.date.DateUtil;
//import cn.hutool.core.util.StrUtil;
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.springsciyon.business.compute.config.LocalCache;
//import com.springsciyon.business.compute.constants.CommonConstant;
//import com.springsciyon.business.compute.dto.HisDataCacheDto;
//import com.springsciyon.business.compute.dto.TdPollData;
//import com.springsciyon.business.compute.dto.TdSubscribeDataDto;
//import com.springsciyon.business.compute.enums.CacheKeyPrefixSuffix;
//import com.springsciyon.business.compute.multidatacource.PointValueConverter;
//import com.springsciyon.business.compute.service.MsgDealService;
//import com.springsciyon.business.compute.util.GeneralFunctionUtil;
//import com.springsciyon.business.compute.util.JedisUtils;
//import com.springsciyon.business.database.v2.model.RtDbConfig;
//import com.springsciyon.business.database.v2.tdengine.TdEngineServer;
//import com.springsciyon.core.tool.utils.CollectionUtil;
//import com.springsciyon.core.tool.utils.SpringUtil;
//import com.taosdata.jdbc.tmq.ConsumerRecord;
//import com.taosdata.jdbc.tmq.ConsumerRecords;
//import com.taosdata.jdbc.tmq.TaosConsumer;
//import org.slf4j.LoggerFactory;
//
//import java.sql.SQLException;
//import java.time.Duration;
//import java.util.*;
//import java.util.stream.Collectors;
//
///**
// * tdengine 订阅模式下 拉取topic测点数据任务
// */
//public class SubscribeDataCacheTask implements Runnable {
//
//    org.slf4j.Logger logger = LoggerFactory.getLogger(getClass());
//
//    /**
//     * tdengine订阅消费者
//     */
//    private TaosConsumer<TdSubscribeDataDto> tdConsumer;
//
//    /**
//     *约定：groupId 由于tdengine的纯数字无法删除groupId，故在TaskVersion前拼接上“group”
//     */
//    private String groupIdOfTdTopic;
//
//    private LocalCache localCache;
//
//    private JedisUtils jedisUtils;
//
//    private RtDbConfig rtDbConfig;
//
//    private MsgDealService msgDealService;
//
//    public SubscribeDataCacheTask(RtDbConfig rtDbConfig) {
//        this.rtDbConfig = rtDbConfig;
//        localCache = SpringUtil.getContext().getBean(LocalCache.class);
//        jedisUtils = SpringUtil.getContext().getBean(JedisUtils.class);
//        groupIdOfTdTopic=(StrUtil.isEmpty(localCache.getTdGroupPreName())?"aalmg":localCache.getTdGroupPreName())+localCache.getTaskVersion();
//        msgDealService = SpringUtil.getContext().getBean(MsgDealService.class);
//    }
//
//    @Override
//    public void run() {
//        try {
//            // TODO: 2025/3/21 测试
//            //等待初次建立报警项关联
//            while(CollectionUtil.isEmpty(localCache.getTagNameSet())){
//                logger.error("订阅：等待任务关联报警项");
//                Thread.sleep(10*1000);
//            }
//            tdConsumer = getTdConsumer(rtDbConfig);
//            startSubscribe(true);
//            pollTargetData(tdConsumer);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    /**
//     * 创建订阅对象
//     * @param uniqueTdDbConfigByEngineSelf
//     * @return
//     * @throws Exception
//     */
//    private TaosConsumer<TdSubscribeDataDto> getTdConsumer(RtDbConfig uniqueTdDbConfigByEngineSelf) throws Exception {
//        Properties config = new Properties();
//        config.setProperty("td.connect.type", "ws");
//        config.setProperty("bootstrap.servers", uniqueTdDbConfigByEngineSelf.getIp()+":"+uniqueTdDbConfigByEngineSelf.getPort());
//        config.setProperty("auto.offset.reset", "latest");
//        config.setProperty("msg.with.table.name", "true");
//        config.setProperty("enable.auto.commit", "true");
//        config.setProperty("auto.commit.interval.ms", "1000");
//        config.setProperty("group.id", groupIdOfTdTopic);
//        config.setProperty("client.id", "aalmc"+localCache.getUuid());
//        config.setProperty("td.connect.user", uniqueTdDbConfigByEngineSelf.getUserName());
//        config.setProperty("td.connect.pass", uniqueTdDbConfigByEngineSelf.getPassword());
//        config.setProperty("value.deserializer", "com.springsciyon.business.compute.dto.TdSubscribeDataDto$ResultDeserializer");
//        config.setProperty("value.deserializer.encoding", "UTF-8");
//
//        try {
//            TaosConsumer<TdSubscribeDataDto> consumer= new TaosConsumer<>(config);
//            logger.error("订阅：构建消费组成功, host: %s, groupId: %s, clientId: %s%n",
//                    config.getProperty("bootstrap.servers"),
//                    config.getProperty("group.id"),
//                    config.getProperty("client.id"));
//            try {
//                logger.error("订阅：构建消费组成功, host: "+config.getProperty("bootstrap.servers")+", groupId:"+config.getProperty("group.id")+", clientId: "+config.getProperty("client.id"));
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//            return consumer;
//        } catch (Exception ex) {
//            try {
//                logger.error("订阅：构建消费组异常, host: "+config.getProperty("bootstrap.servers")+", groupId:"+config.getProperty("group.id")+", clientId: "+config.getProperty("client.id"));
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//            // please refer to the JDBC specifications for detailed exceptions info
//            logger.error("订阅：构建消费组异常, host: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
//                    config.getProperty("bootstrap.servers"),
//                    config.getProperty("group.id"),
//                    config.getProperty("client.id"),
//                    ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
//                    ex.getMessage());
//            // Print stack trace for context in examples. Use logging in production.
//            ex.printStackTrace();
//            throw ex;
//        }
//    }
//
//
//    /**
//     * 建立订阅连接，监测redis是否需要重置订阅
//     * @param init
//     * @return
//     * @throws SQLException
//     * @auth xty
//     */
//    private boolean startSubscribe (boolean init) throws SQLException {
//        boolean reset=false;
//        String backWriteKey = jedisUtils.get(CacheKeyPrefixSuffix.TASK_RESET_SUBSCRIBE_FRONT + localCache.getTaskVersion());
//        //如果通用服务反写测点标签需要重置
//        if(StrUtil.isNotEmpty(backWriteKey)&& CommonConstant.GENERAL_BACK_WRITE.equals(backWriteKey)){
//            reset=true;
//            jedisUtils.set(CacheKeyPrefixSuffix.TASK_RESET_SUBSCRIBE_FRONT + localCache.getTaskVersion(),CommonConstant.COMPUTE_RESET);
//        }
//        //逗号分隔，支持同时订阅多个topic(因为是订阅超表，bool和float类型都属不同的topic)
//        String tdTopic = localCache.getTdTopic();
//        List<String> topics = Arrays.stream(tdTopic.split(","))
//                .collect(Collectors.toList());
//        if(reset){
//            if(!init){
//                tdConsumer.();
//            }
//            //如果订阅进行测点删除,可能会存在测点遗漏,如果不删除会导致反写标签的数据无法被订阅到
//            Boolean aBooleanOfDropGroup = funOfDropGroup(groupIdOfTdTopic, topics);
//            if(!aBooleanOfDropGroup){
//                logger.error("订阅：存在订阅前先删除绑定得组id失败");
//            }
//            tdConsumer.subscribe(topics);
//            if(localCache.getTaskStat()){
//                if(init){
//                    logger.error("订阅：初始化重置数据订阅");
//                }else{
//                    logger.error("订阅：反写标签重置数据订阅");
//                }
//            }
//        }else{
//            if(init){
//                tdConsumer.subscribe(topics);
//                if(localCache.getTaskStat()) {
//                    logger.error("订阅：初始化数据订阅");
//                }
//            }
//        }
//        return reset;
//    }
//
//    /**
//     * 拉取数据
//     * 数据可能时间不通
//     * @param consumer
//     */
//    public void pollTargetData(TaosConsumer<TdSubscribeDataDto> consumer) {
//
//        //用于存储拉取测点数据，不同测点的正向时间序列集合映射
//        Map<String, List<HisDataCacheDto>> realMapByTagName = new HashMap<>();
//
//        long startTime=System.currentTimeMillis();
//        int count=0;
//        //todo 额外测试
//        //轮询
//        while (true) {
//            try {
//                //如果没有测点数据就不拉取数据
//                if(CollectionUtil.isNotEmpty(localCache.getTagNameSet())) {
//                    // poll data
//                    ConsumerRecords<TdSubscribeDataDto> records = consumer.poll(Duration.ofMillis(1000));
//
//                    //把拉取到的数据打印
//                    // 打印 变量控制
//                    try {
//                        if(localCache.getPollTdTagLog()){
//                            List<TdPollData> tempList=new ArrayList<>();
//                            for (ConsumerRecord<TdSubscribeDataDto> record : records) {
//                                TdSubscribeDataDto value = record.value();
//                                TdPollData tdPollData = buildTdPollData(value);
//                                tempList.add(tdPollData);
//                            }
//                            logger.error("订阅poll到："+ JSON.toJSONString(tempList));
//                            logger.error("records.isEmpty()的标志位:"+records.isEmpty());
//                            logger.error("拉完处理之前"+JSON.toJSONString(realMapByTagName));
//                        }
//                    } catch (Exception e) {
//                        e.printStackTrace();
//                    }
//                    //若拉取不到数据直接下一次拉取
//                    if (!records.isEmpty()) {
//                        if(localCache.getPollTdTagLog()){
//                            logger.error("进入外层1");
//                        }
//                        for (ConsumerRecord<TdSubscribeDataDto> record : records) {
//                            TdSubscribeDataDto value = record.value();
//                            String tbName = value.getTbname();
//                            HisDataCacheDto hisDataCacheDto = buildHisDataCacheDto(value);
//
//                            if (localCache.getTagNameSet().contains(tbName)) {
//                                if(localCache.getPollTdTagLog()){
//                                    logger.error("测点名包含"+tbName);
//                                }
//                                try {
//                                    if (realMapByTagName.containsKey(tbName)) {
//                                        if(localCache.getPollTdTagLog()){
//                                            logger.error("realMapByTagName已包含"+tbName);
//                                        }
//                                        List<HisDataCacheDto> dataCacheDtoList = realMapByTagName.get(tbName);
//                                        //正常情况不会走入的这个逻辑，所以效率不受影响,但是td数据拉取不能保证会不会拉取到相同测点数据
//                                        int i = 0;
//                                        boolean isNotInsert = false;
//                                        for (HisDataCacheDto dataCacheDto : dataCacheDtoList) {
//                                            if (dataCacheDto.getDateTime().compareTo(hisDataCacheDto.getDateTime()) == 0) {
//                                                if (localCache.getPollTdTagLog()) {
//                                                    logger.error("compareTo执行了"+tbName);
//                                                }
//                                                isNotInsert = true;
//                                                if (!dataCacheDto.getHisCacheData().equals(hisDataCacheDto.getHisCacheData())) {
//                                                    if(localCache.getTaskStat()){
//                                                        logger.error("订阅："+JSONObject.toJSONString(hisDataCacheDto) + "内容不一致" + JSONObject.toJSONString(dataCacheDto));
//                                                    }
//                                                    if (localCache.getPollTdTagLog()) {
//                                                        logger.error("equals执行了"+tbName);
//                                                    }
//                                                    break;
//                                                }
//                                                break;
//                                            } else if (dataCacheDto.getDateTime().compareTo(hisDataCacheDto.getDateTime()) > 0) {
//                                                if (localCache.getPollTdTagLog()) {
//                                                    logger.error("大于0执行了"+tbName);
//                                                }
//                                                break;
//                                            }
//                                            i++;
//                                        }
//                                        //保持正向有序数列
//                                        if (!isNotInsert) {
//                                            if (localCache.getPollTdTagLog()) {
//                                                logger.error("isNotInsert执行了"+tbName);
//                                            }
//                                            if(localCache.getTaskStat()) {
//                                                logger.error("订阅：拉取同一测点不同时间数据"+JSONObject.toJSONString(hisDataCacheDto));
//                                            }
//                                            dataCacheDtoList.add(i, hisDataCacheDto);
//                                        }
//                                    } else {
//                                        try {
//                                            if (localCache.getPollTdTagLog()) {
//                                                logger.error("先realMapByTagName"+JSON.toJSONString(realMapByTagName));
//                                            }
//                                        } catch (Exception e) {
//                                            e.printStackTrace();
//                                        }
//                                        realMapByTagName.put(tbName, new ArrayList<>());
//                                        realMapByTagName.get(tbName).add(hisDataCacheDto);
//                                        try {
//                                            if (localCache.getPollTdTagLog()) {
//                                                logger.error("后realMapByTagName"+JSON.toJSONString(realMapByTagName));
//                                            }
//                                        } catch (Exception e) {
//                                            e.printStackTrace();
//                                        }
//                                    }
//                                } catch (Exception e) {
//                                    e.printStackTrace();
//                                }
//                            }
//                        }
//                    }
//                    //为了保证和数据刷新尽可能同频避免队列溢出
//                    if (System.currentTimeMillis() - startTime > 1000) {
//                        if (localCache.getPollTdTagLog()) {
//                            logger.error("执行DataPush");
//                        }
//                        startTime = DataPush(realMapByTagName);
//                        count += 1;
//                        //每过五分钟，检测是否被重写了，因为反写标签涉及报警项初次构建，所以5min是可以容忍
//                        if (count > 300) {
//                            count = 0;
//                            if (localCache.getPollTdTagLog()) {
//                                logger.error("执行startSubscribe");
//                            }
//                            startSubscribe(false);
//                        }
//
//                    }else{
//                        try {
//                            if (localCache.getPollTdTagLog()) {
//                                logger.error("未达到1000ms"+JSON.toJSONString(realMapByTagName));
//                            }
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                        }
//                    }
//                }else{
//                    try {
//                        logger.error("订阅：等待报警项数据");
//                        Thread.sleep(60 * 1000);
//                    } catch (InterruptedException e) {
//                        // 处理中断异常
//                        e.printStackTrace();
//                    }
//                }
//            } catch (Exception ex) {
//                logger.error("订阅：拉取数据异常, topic: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
//                        localCache.getTdTopic(),
//                        groupIdOfTdTopic,
//                        localCache.getUuid(),
//                        ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
//                        ex.getMessage());
//                ex.printStackTrace();
//            }
//        }
//    }
//
//    /**
//     * 订阅数据推送处理
//     * @param realMapByTagName
//     * @return
//     * @auth xty
//     */
//    private long DataPush(Map<String, List<HisDataCacheDto>> realMapByTagName){
//
//        // 打印 变量控制
//        if(localCache.getSubTdTagLog()){
//            logger.error("订阅：DataPush"+ JSON.toJSONString(realMapByTagName));
//        }
//
//        if (localCache.getBackWriteTdTag()) {
//            long b1 = System.nanoTime();
//            msgDealService.saveTdRecords(realMapByTagName);
//            logger.error("saveTdRecords返写td"+realMapByTagName.size()+"个返回值，耗时："+(System.nanoTime()-b1)/1000000+"毫秒");
//        }
//
//        //todo 额外测试
//        if(CollectionUtil.isNotEmpty(realMapByTagName)) {
//            boolean offer = localCache.getTdSubscribeDataQueue().offer(new HashMap<>(realMapByTagName));
//            if (!offer) {
//                logger.error("订阅：TdSubscribeDataQueue队列已满");
//                String val="C"+new DateTime().getTime() +";";
//                GeneralFunctionUtil.timeoutRecords("subscribe",localCache.getTaskVersion(),val,jedisUtils);
//            }else{
//                if(localCache.getTaskStat()){
//                    logger.error("订阅：推送成功"+realMapByTagName.size());
//                }
//            }
//            realMapByTagName.clear();
//        }
//        return System.currentTimeMillis();
//    }
//    /**
//     * 删除数据订阅的消费者组,如果不做删除处理可能会存在数据反写标签检测不到数据
//     * @param groupId
//     * @param topics
//     * @return
//     */
//    public Boolean funOfDropGroup(String groupId, List<String> topics){
//        TdEngineServer tdEngineServer = new TdEngineServer();
//        tdEngineServer.init(rtDbConfig);
//        Boolean resFlag=true;
//        for (String topic : topics) {
//            Boolean aBoolean = tdEngineServer.dropConsumer(groupId, topic);
//            if(resFlag&& (!aBoolean)){
//                resFlag=false;
//            }
//        }
//        return resFlag;
//    }
//
//    /**
//     * 获取订阅实体对象
//     * @param value
//     * @return
//     */
//    public HisDataCacheDto buildHisDataCacheDto(TdSubscribeDataDto value) {
//        HisDataCacheDto hisDataCacheDto = new HisDataCacheDto();
//        Object v = value.getV();
//        DateTime dateTime = DateUtil.date(value.getTs());
//        hisDataCacheDto.setDateTime(dateTime);
//        hisDataCacheDto.setHisCacheData(PointValueConverter.getNum(String.valueOf(v)));
//        return hisDataCacheDto;
//    }
//
//    public TdPollData buildTdPollData(TdSubscribeDataDto value) {
//        TdPollData tdPollData = new TdPollData();
//        Object v = value.getV();
//        DateTime dateTime = DateUtil.date(value.getTs());
//        tdPollData.setDateTime(dateTime);
//        tdPollData.setHisCacheData(PointValueConverter.getNum(String.valueOf(v)));
//        tdPollData.setTbName(value.getTbname());
//        return tdPollData;
//    }
//
//
//}
